package cn.lgwen.candy.connector.kafka010.sink;

import cn.lgwen.candy.runtime.operator.SinkOperator;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 2019/12/26
 * aven.wu
 * danxieai258@163.com
 */
public class KafkaSink010<V> extends SinkOperator<V> {

    private String topic;

    private Properties kafkaProperties;

    private KafkaProducer<byte[], V> kafkaProducer;

    private KeySerialize<V> keySerialize;

    public KafkaSink010(String topic, Properties kafkaProperties) {
        this.topic = topic;
        this.kafkaProperties = kafkaProperties;
        kafkaProducer = new KafkaProducer<>(kafkaProperties);
    }

    public KafkaSink010(String topic, Properties kafkaProperties, KeySerialize<V> keySerialize) {
        this.topic = topic;
        this.kafkaProperties = kafkaProperties;
        kafkaProducer = new KafkaProducer<>(kafkaProperties);
        this.keySerialize = keySerialize;
    }

    @Override
    public void invoke(V input) {
        ProducerRecord<byte[], V> record =
                new ProducerRecord<>(topic, keySerialize.keySerialize(input), input);
        kafkaProducer.send(record);
    }
}
